package com.sui.bigdata.sml.web.misc.websocket;

import com.feidee.fdcommon.azkaban.constant.SparkConstant;
import com.feidee.fdcommon.configuration.CustomConfiguration;

import javax.websocket.Session;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

/**
 * @Author songhaicheng
 * @Date 2019/8/26 19:42
 * @Description
 * @Reviewer
 */
public class NodeLogClient extends Thread {

    private String nid;
    private Session session;
    private String logPath;
    private volatile boolean stop;

    public String getNid() {
        return nid;
    }

    public void setNid(String nid) {
        this.nid = nid;
    }

    public Session getSession() {
        return session;
    }

    public void setSession(Session session) {
        this.session = session;
    }

    public String getLogPath() {
        return logPath;
    }

    public void setLogPath(String logPath) {
        this.logPath = logPath;
    }


    public NodeLogClient(String nid, Session session) {
        this.nid = nid;
        this.session = session;
        this.logPath = CustomConfiguration.getString(SparkConstant.SML_EXECUTION_LOG_DIR) + "/" + nid + ".log";
        stop = false;
    }

    void stopPushingLog() {
        this.stop = true;
    }

    @Override
    public int hashCode() {
        return (nid + "_" + session.getId()).hashCode();
    }

    @Override
    public boolean equals(Object o) {
        if (o == null) {
            return false;
        }
        if (this == o) {
            return true;
        }
        if (o instanceof NodeLogClient) {
            return this.hashCode() == o.hashCode();
        }
        return false;
    }

    @Override
    public void run() {
        FileInputStream fin = null;
        try {
            File file = new File(logPath);
            if (!file.exists()) {
                session.getBasicRemote().sendText("<font color='red'>日志文件不存在！！！</font>");
            } else {
                fin = new FileInputStream(file);
                FileChannel channel = fin.getChannel();
                int capacity = 1024;
                ByteBuffer bf = ByteBuffer.allocate(capacity);
                int length;
                // -2 代表一直监听新的日志内容
                while ((length = channel.read(bf)) != -2 && !stop) {
                    if (length != -1) {
                        bf.clear();
                        byte[] bytes = bf.array();
                        session.getBasicRemote().sendText(new String(bytes, 0, length));
                    }
                }
                channel.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (fin != null) {
                try {
                    fin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
